package com.google.code.mochaccino.framework.jms;
/*
 * Copyright 2012 Claude Houle claude.houle@gmail.com
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.google.common.base.Preconditions;
import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import javax.inject.Inject;
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** The Message Broker Client allows a end-user to interface with a JMS-compliant message broker. */
public final class JmsClient {

	/** Logger */
	private static final Logger logger = LoggerFactory.getLogger( JmsClient.class );

	/** The Connection Factory */
	private ConnectionFactory factory = null;

	/** The Queue Connection */
	private Connection connection;

	/** The Destination */
	private Destination destination;

	/** Session */
	private Session session = null;

	/** Destination Name */
	private String destinationName;

	/** Is Queue */
	private boolean queue;

	/** Transacted */
	private boolean transacted;

	/** Default Delivery Mode */
	private JmsDeliveryMode defaultDeliveryMode = JmsDeliveryMode.PERSISTENT;

	/** Constructor */
	public JmsClient( ConnectionFactory factory, String name, boolean isQueue, JmsDeliveryMode mode ) {
		this( factory, name, true, isQueue, mode );
	}

	/** Constructor */
	@Inject
	public JmsClient( ConnectionFactory factory, String name, boolean transacted, boolean isQueue, JmsDeliveryMode mode ) {
		this.factory = factory;
		this.destinationName = name;
		this.transacted = transacted;
		this.queue = isQueue;
		this.defaultDeliveryMode = mode;
	}

	/** Close the Message Broker Client */
	public boolean close() {
		if ( !isStarted() ) {
			return false;
		}

		return closeQuietly();
	}

	/** Close Every Resource Quietly */
	private boolean closeQuietly() {
		boolean success = true;
		if ( !JmsUtilities.closeQuietly( session ) ) {
			success = false;
		}

		if ( !JmsUtilities.closeQuietly( connection ) ) {
			success = false;
		}

		this.connection = null;
		this.session = null;
		this.destination = null;
		return success;
	}

	/** Utility method to create a message */
	public BytesMessage createBytes() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		BytesMessage msg = session.createBytesMessage();
		customizeMessage( msg );
		return msg;
	}

	private void customizeMessage( Message msg ) throws JMSException {
		msg.setJMSPriority( Message.DEFAULT_PRIORITY );
		msg.setJMSDeliveryMode( DeliveryMode.PERSISTENT );
		msg.setJMSExpiration( -1 );
		msg.setJMSTimestamp( System.currentTimeMillis() );
	}

	/** Utility method to create a message */
	public BytesMessage createBytes( byte[] payload ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		BytesMessage msg = session.createBytesMessage();
		msg.writeBytes( Preconditions.checkNotNull( payload ) );
		customizeMessage( msg );
		return msg;
	}

	/** Utility method to create a message */
	public MapMessage createMapMessage() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		MapMessage msg = session.createMapMessage();
		customizeMessage( msg );
		return msg;
	}

	/** Utility method to create a message */
	public MapMessage createMapMessage( Map<String,Serializable> payload ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		MapMessage msg = session.createMapMessage();
		if ( payload != null ) {
			for ( Map.Entry<String,Serializable> entry : payload.entrySet() ) {
				msg.setObject( entry.getKey(), entry.getValue() );
			}
		}
		customizeMessage( msg );
		return msg;
	}

	/** Utility method to create a message */
	public ObjectMessage createObject() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		ObjectMessage msg = session.createObjectMessage();
		customizeMessage( msg );
		return msg;
	}

	/** Utility method to create a message */
	public ObjectMessage createObject( Serializable payload ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		ObjectMessage msg = session.createObjectMessage( payload );
		customizeMessage( msg );
		return msg;
	}

	/** Utility method to create a message */
	public StreamMessage createStream() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		StreamMessage msg = session.createStreamMessage();
		return msg;
	}

	/** Utility method to create a message */
	public StreamMessage createStream( InputStream payload ) throws JMSException, IOException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		StreamMessage msg = session.createStreamMessage();
		int b = -1;
		while ( (b = payload.read()) != -1 ) {
			msg.writeByte( (byte) b );
		}
		customizeMessage( msg );
		return msg;
	}

	/** Utility method to create a message */
	public TextMessage createText() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		TextMessage msg = session.createTextMessage();
		customizeMessage( msg );
		return msg;
	}

	/** Utility method to create a message */
	public TextMessage createText( String payload ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		TextMessage msg = session.createTextMessage( payload );
		customizeMessage( msg );
		return msg;
	}

	/** Is Started? */
	public boolean isStarted() {
		return session != null && destination != null && connection != null;
	}

	/** Start the Message Broker Client */
	public boolean open() {
		return open( JmsSessionMode.AUTO );
	}

	/** Start the Message Broker Client */
	public boolean open( JmsSessionMode mode ) {
		if ( isStarted() ) {
			return false;
		}

		try {
			this.connection = this.factory.createConnection();
			this.connection.start();
			this.session = this.connection.createSession( transacted, mode.getMode() );
			this.destination = queue ? this.session.createQueue( destinationName ) : this.session.createTopic( destinationName );
			return true;
		} catch ( JMSException e ) {
			closeQuietly();
			logger.error( e.getMessage(), e );
			return false;
		}
	}

	/** Post messages */
	public int post( Message... messages ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		return post( destination, messages );
	}

	/** Post messages */
	private int post( Destination alternateDestination, Message... messages ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}

		int count = 0;
		if ( messages != null ) {
			MessageProducer sender = null;
			try {
				sender = createMessageProducer();
				for ( Message message : messages ) {
					sender.send( alternateDestination, message );
					count++;
				}
			} finally {
				JmsUtilities.closeQuietly( sender );
			}
		}
		return count;
	}

	/** Create Message Producer */
	private MessageProducer createMessageProducer() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}

		MessageProducer producer = this.session.createProducer( destination );
		producer.setDeliveryMode( defaultDeliveryMode.getMode() );
		return producer;
	}

	/** Post messages */
	public int post( String destinationName, boolean isQueue, Message... messages ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}

		return post( isQueue ? session.createQueue( destinationName ) : session.createTopic( destinationName ), messages );
	}

	/** Receive a message and wait infinity */
	public Message receive() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		MessageConsumer consumer = null;
		try {
			consumer = createMessageConsumer();
			return consumer.receive();
		} finally {
			JmsUtilities.closeQuietly( consumer );
		}
	}

	/** Create Message Consumer */
	private MessageConsumer createMessageConsumer() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		return this.session.createConsumer( destination );
	}

	/** Receive a message and wait specified time */
	public Message receive( TimeUnit unit, long timeout ) throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		MessageConsumer consumer = null;
		try {
			consumer = createMessageConsumer();
			return consumer.receive( unit.toMillis( timeout ) );
		} finally {
			JmsUtilities.closeQuietly( consumer );
		}
	}

	/** Receive a message without wait */
	public Message receiveNoWait() throws JMSException {
		if ( !isStarted() ) {
			throw new IllegalStateException( "Not Started!" );
		}
		MessageConsumer consumer = null;
		try {
			consumer = createMessageConsumer();
			return consumer.receiveNoWait();
		} finally {
			JmsUtilities.closeQuietly( consumer );
		}
	}
}
